package zkclient

import (
	"EasyData/global"
	"EasyData/util"
	"github.com/samuel/go-zookeeper/zk"
	"time"
	"fmt"
	"EasyData/monitor"
)

const (
	rootNode= "/easydata" //根节点
	mysql="mysql"//mysql配置节点名
	goroutine="goroutine"//协程数
	memory="memory"//内存
	hbase="hbase"//Hbase配置节点的节点名
	taskCommond ="taskCommond"//任务描述
	speed="speed"//任务限速
	onlyDelete="onlyDelete"//是否是onlydetete任务，是的话，则查询出数据就删除，不往Hbase中导入
	dataBase="dataBase"//数据库名
	table="table"//表名
	maxOpenConns="maxOpenConns"//最大连接数
	maxIdleConns="maxIdleConns"//最大空闲连接
	key="key"//数据库主键
	condition="condition"//条件
	parameter="parameter"//参数类型
	cron="cron"//定时任务表达式
	isFlow="isFlow"//是否是流式任务
	limit="limit"//每次查询的结果集个数
	deleter="delete"//数据同步成功后，是否Delete掉原数据
	syncColumn="syncColumn"//当只是同步数据到Hbase，并不删除源数据，需要提供一个标示是否已经同步的字段
	syncValue="syncValue"//当只是同步数据到Hbase，并不删除源数据，则同步成功后，将sync_column对应的字段改为sync_value
	hbaseTable="hbaseTable"//hbase数据库的表名
	separator="Separator"//rowkey中多个字段的分割符
	rowKey="rowKey"//rowkey的字段列表，用,号间隔
	partOf="partOf"//是否允许只导入部分的列到Hbase，配置成true表示允许，否则，不允许
	toLower="toLower"//是否列名在Hbase中转小写
	toUpper="toUpper"//是否列名在Hbase中转大写
	cloumnFamily="cloumnFamily"//列族配置
	open="open"//一个开关，用户从zk上手动触发修改为任意值后，手动触发定时任务
	stop="stop"//用于对正在执行的定时任务进行暂停和启动

)

/**
	开始配置每一个任务的具体配置项目，并注册监听
 */
func createChildSessionNodeIfExists(task *global.Task) {

    go createIfNotExists(task)

	//针对某些热配置有意义的属性，进行了监听和回调
	/**
		限制需要人工触发的任务的速率
	 */
	watchSpeed := &taskWatcherAndNotify{make(chan (<-chan zk.Event)),task,getTaskConfigNode(task.TaskName,speed)}
	watchSpeed.StartWatchAndNotify()

	/**
		支持在线修改表名
	 */
	WatchTable := &taskWatcherAndNotify{make(chan (<-chan zk.Event)),task,getTaskConfigNode(task.TaskName,table)}
	WatchTable.StartWatchAndNotify()
	/**
		支持在线修改主键
	 */
	WatchKey := &taskWatcherAndNotify{make(chan (<-chan zk.Event)),task,getTaskConfigNode(task.TaskName,key)}
	WatchKey.StartWatchAndNotify()
	/**
		支持在线修改表达式
	 */
	WatchCondition :=&taskWatcherAndNotify{make(chan (<-chan zk.Event)),task,getTaskConfigNode(task.TaskName,condition)}
	WatchCondition.StartWatchAndNotify()
	/**
		支持在线修改表达式参数
	 */
	WatcherParameter :=&taskWatcherAndNotify{make(chan (<-chan zk.Event)),task,getTaskConfigNode(task.TaskName,parameter)}
	WatcherParameter.StartWatchAndNotify()
	/**
		支持在线修改任务的查询条数
	 */
	WatcherLimit :=&taskWatcherAndNotify{make(chan (<-chan zk.Event)),task,getTaskConfigNode(task.TaskName,limit)}
	WatcherLimit.StartWatchAndNotify()
	/**
		支持在线修改 同步字段的字段名
	 */
	WatcherSyncColumn :=&taskWatcherAndNotify{make(chan (<-chan zk.Event)),task,getTaskConfigNode(task.TaskName,syncColumn)}
	WatcherSyncColumn.StartWatchAndNotify()
	/**
		支持在线修改 同步字段的值，即 当数据导入Hbase中后，将mysql中的SyncColumn 改为SyncValue
	 */
	WatcherSyncValue :=&taskWatcherAndNotify{make(chan (<-chan zk.Event)),task,getTaskConfigNode(task.TaskName,syncValue)}
	WatcherSyncValue.StartWatchAndNotify()
	/**
		手动触发定时任务
	 */
	WatcherOpen :=&taskWatcherAndNotify{make(chan (<-chan zk.Event)),task,getTaskConfigNode(task.TaskName,open)}
	WatcherOpen.StartWatchAndNotify()
	/**
		暂停定时任务
	 */
	WatcherStop :=&taskWatcherAndNotify{make(chan (<-chan zk.Event)),task,getTaskConfigNode(task.TaskName,stop)}
	WatcherStop.StartWatchAndNotify()

}
/**
	如果配置节点不存在，就创建
	采用心跳检测的方式,原因：程序重启期间，zk会删除临时节点，因此，只创建一次节点可能会失效
 */
func createIfNotExists(task *global.Task) {
	for {
		createSessionNodeIfExists(getTaskConfigNode(task.TaskName,taskCommond),util.ToString(task.TaskCommond))
		createSessionNodeIfExists(getTaskConfigNode(task.TaskName,speed),util.ToString(task.Speed))
		createSessionNodeIfExists(getTaskConfigNode(task.TaskName,onlyDelete),util.ToString(task.OnlyDelete))
		createSessionNodeIfExists(getTaskConfigNode(task.TaskName,dataBase),util.ToString(task.Database))
		createSessionNodeIfExists(getTaskConfigNode(task.TaskName,table),util.ToString(task.Table))
		createSessionNodeIfExists(getTaskConfigNode(task.TaskName,maxOpenConns),util.ToString(task.MaxOpenConns))
		createSessionNodeIfExists(getTaskConfigNode(task.TaskName,maxIdleConns),util.ToString(task.MaxIdleConns))
		createSessionNodeIfExists(getTaskConfigNode(task.TaskName,key),util.ToString(task.Key))
		createSessionNodeIfExists(getTaskConfigNode(task.TaskName,condition),util.ToString(task.Condition))
		createSessionNodeIfExists(getTaskConfigNode(task.TaskName,parameter),util.ToString(task.Parameter))
		createSessionNodeIfExists(getTaskConfigNode(task.TaskName,cron),util.ToString(task.Cron))
		createSessionNodeIfExists(getTaskConfigNode(task.TaskName,isFlow),util.ToString(task.IsFlow))
		createSessionNodeIfExists(getTaskConfigNode(task.TaskName,limit),util.ToString(task.Limit))
		createSessionNodeIfExists(getTaskConfigNode(task.TaskName,deleter),util.ToString(task.Delete))
		createSessionNodeIfExists(getTaskConfigNode(task.TaskName,syncColumn),util.ToString(task.SyncColumn))
		createSessionNodeIfExists(getTaskConfigNode(task.TaskName,syncValue),util.ToString(task.SyncValue))
		createSessionNodeIfExists(getTaskConfigNode(task.TaskName,hbaseTable),util.ToString(task.HbaseTable))
		createSessionNodeIfExists(getTaskConfigNode(task.TaskName,separator),util.ToString(task.Separator))
		createSessionNodeIfExists(getTaskConfigNode(task.TaskName,rowKey),util.ToString(task.RowKey))
		createSessionNodeIfExists(getTaskConfigNode(task.TaskName,partOf),util.ToString(task.PartOf))
		createSessionNodeIfExists(getTaskConfigNode(task.TaskName,toLower),util.ToString(task.ToLower))
		createSessionNodeIfExists(getTaskConfigNode(task.TaskName,toUpper),util.ToString(task.ToUpper))
		createSessionNodeIfExists(getTaskConfigNode(task.TaskName,cloumnFamily),util.ToString(task.CloumnFamily))
		createSessionNodeIfExists(getTaskConfigNode(task.TaskName,open),"123456")
		createSessionNodeIfExists(getTaskConfigNode(task.TaskName,stop),"false")

		time.Sleep(30*time.Second)
	}
}
/**
	在应用的namespace下创建协程数量和内存利用率的信息
 */
func createIfNoExistsMemAndGetGoroutine() {
	for{
		createSetSessionNode(getTaskNode(goroutine),fmt.Sprintf("%v",monitor.GetGoroutine()))
		v1,v2,v3,v4:= monitor.GetMemoryMb()
		createSetSessionNode(getTaskNode(memory),fmt.Sprintf("%s,%s,%s,%s",v1,v2,v3,v4))

		time.Sleep(10*time.Second)
	}
}
